package bma.common.netty.thrift.client;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.thrift.TBase;
import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransport;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.util.internal.ConcurrentHashMap;

import bma.common.langutil.ai.stack.AIStack;
import bma.common.netty.SupportedNettyChannel;
import bma.common.netty.client.CommonNettyClientSender;
import bma.common.netty.client.NettyClient;
import bma.common.netty.client.NettyClientCloseListener;
import bma.common.netty.client.NettyClientExceptionListener;
import bma.common.netty.client.NettyClientMessageListener;
import bma.common.netty.thrift.core.NCHFramed;
import bma.common.netty.thrift.core.TNettyChannelBufferTransport;
import bma.common.netty.thrift.core.TNettyFramedWriteOnlyTransport;
import bma.common.thrift.ThriftClientConfig;
import bma.common.thrift.ai.AIThriftInvoker;
import bma.common.thrift.ai.TAIBaseServiceClient;

public class AIThriftInvokerNettySocket implements AIThriftInvoker,
		SupportedNettyChannel {

	protected NettyClient client;
	protected ThriftClientConfig config;
	protected Map<Integer, AIStack<?>> waiting;

	public AIThriftInvokerNettySocket(NettyClient client, ThriftClientConfig cfg) {
		super();
		this.config = cfg;
		setClient(client);
	}

	public NettyClient getClient() {
		return client;
	}

	public void setClient(NettyClient client) {
		bindClient(client);
	}

	public void bindClient(NettyClient client) {
		this.client = client;
		if (config.isNoneBlock()) {
			if (client.missHandler("frame")) {
				client.addHandler("frame",
						new NCHFramed(config.getFrameMaxLength()));
			}
			client.addListener(new Listener());
			if (waiting != null) {
				waiting = new ConcurrentHashMap<Integer, AIStack<?>>();
			}
		}
	}

	@Override
	public Channel getChannel() {
		return this.client != null ? this.client.getChannel() : null;
	}

	@Override
	public <TYPE> boolean invoke(AIStack<TYPE> stack, int seqid, TBase result,
			String moduleName, String name, TBase avgs) throws TException {
		if (config.isNoneBlock()) {
			return noneblockInvoke(stack, seqid, result, moduleName, name, avgs);
		} else {
			return blockInvoke(stack, seqid, result, moduleName, name, avgs);
		}
	}

	public <TYPE> boolean blockInvoke(AIStack<TYPE> stack, final int seqid,
			final TBase result, final String moduleName, final String name,
			final TBase avgs) throws TException {
		client.send(new CommonNettyClientSender<TYPE>(stack) {

			@Override
			public void bindNettyClient(NettyClient client, boolean bind) {
				if (bind) {
					client.addHandler("frame",
							new NCHFramed(config.getFrameMaxLength()));
				} else {
					client.removeHandler("frame");
				}
			}

			@Override
			public boolean messageReceived(final NettyClient client,
					Object message) {
				if (message instanceof ChannelBuffer) {
					ChannelBuffer cb = (ChannelBuffer) message;
					TTransport transport = new TNettyChannelBufferTransport(cb);
					try {
						TAIBaseServiceClient.readBase(subStack(client),
								config.createProtocol(transport, moduleName),
								seqid, name, result);
					} catch (Exception e) {
						end(client, null, e);
					}
					return true;
				}
				return false;
			}

			@Override
			public void sendRequest(NettyClient client) {
				try {
					TTransport transport = new TNettyFramedWriteOnlyTransport(
							client.getChannel(), config.getFrameMaxLength());
					TAIBaseServiceClient.sendBase(
							config.createProtocol(transport, moduleName), name,
							avgs, seqid);
					if (result == null) {
						end(client, null, null);
					}
				} catch (Exception e) {
					end(client, null, e);
				}
			}
		});
		return false;
	}

	protected class Listener implements NettyClientExceptionListener,
			NettyClientCloseListener, NettyClientMessageListener {

		@Override
		public boolean messageReceived(NettyClient client, Object message) {
			if (message instanceof ChannelBuffer) {
				ChannelBuffer cb = (ChannelBuffer) message;
				TTransport transport = new TNettyChannelBufferTransport(cb);
//				try {
//					TAIBaseServiceClient.readBase(subStack(client),
//							config.createProtocol(transport, moduleName),
//							seqid, name, result);
//				} catch (Exception e) {
//					end(client, null, e);
//				}
//				return true;
			}
			return false;
		}

		@Override
		public void clientClose(NettyClient client, boolean remoteClose) {
			if (remoteClose) {
				handleError(new IOException("closed"));
			}
		}

		@Override
		public boolean exceptionCaught(NettyClient client, Throwable t) {
			handleError(t);
			return true;
		}

		public void handleError(Throwable t) {
			List<AIStack<?>> list = new ArrayList<AIStack<?>>(waiting.values());
			
		}

	}

	public <TYPE> boolean noneblockInvoke(AIStack<TYPE> stack, final int seqid,
			final TBase result, final String moduleName, final String name,
			final TBase avgs) throws TException {
		try {
			TTransport transport = new TNettyFramedWriteOnlyTransport(
					client.getChannel(), config.getFrameMaxLength());
			TAIBaseServiceClient.sendBase(
					config.createProtocol(transport, moduleName), name, avgs,
					seqid);
			if (result == null) {
				return stack.success(null);
			} else {
				waiting.put(seqid, stack);
				return false;
			}
		} catch (Exception e) {
			return stack.failure(e);
		}
	}
}
